package com.blue.rpc.server;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.task.TaskExecutor;

import com.blue.rpc.core.annotation.RpcParser;
import com.blue.rpc.core.invoke.RpcInvoker;
import com.blue.rpc.core.message.RpcRequest;
import com.blue.rpc.core.message.RpcResponse;
import com.blue.rpc.util.RpcConst;
import com.blue.rpc.util.RpcServerUtil;

import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * RPC远程调用处理器
 * 
 * @author zhengj
 * @since 1.0 2017年3月13日
 */
public class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest>
{
	private static Logger logger = LoggerFactory.getLogger(RpcServerHandler.class);
	
	private final TaskExecutor taskExecutor;
	private final RpcInvoker invoker;
	private final String token;

	public RpcServerHandler(TaskExecutor taskExecutor, RpcParser parser, String token)
	{
		this.taskExecutor = taskExecutor;
		this.token = token;
		this.invoker = new RpcInvoker(parser);
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception
	{
		super.channelActive(ctx);
		logger.info("客户端连接：{}", ctx.channel().remoteAddress());
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception
	{
		super.channelInactive(ctx);
		logger.info("客户端断开连接：{}", ctx.channel().remoteAddress());
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception
	{
		logger.error("有错误－", cause);
		ctx.close();
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request)
			throws Exception
	{
		if (request.getType() == RpcConst.SHAKE) // 握手报文
		{
			RpcResponse response = RpcServerUtil.create(request);
			response.setRequestId(token);
			if (request.getRequestId() != null && request.getRequestId().equals(token)) // 验证成功
			{
				logger.info("握手成功，客户端：{}，token：{}", ctx.channel().remoteAddress(), token);
				ctx.writeAndFlush(response);
			}
			else
			{
				logger.warn("握手失败，客户端：{}，本地 token：{}，远程 token：{}", ctx.channel().remoteAddress(), token, request.getRequestId());
				ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
			}
		}
		else
		{
			taskExecutor.execute(() ->
			{
				RpcResponse response = invoker.invoke(request);
				ctx.writeAndFlush(response);
			});
		}
	}

}
